package com.ww.springboot.boot.rxjava;

import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

/**
 * 描述：
 *
 * @author 🧑 ‍wanwei
 * @since 2022-04-25 16:35
 */
public class RxJavaDemo1 {

    public static void main(String[] args) {
        //被观察者
        Observable observable = Observable.create(e -> {
            System.out.println(("=========================currentThread name able: " + Thread.currentThread().getName()));
            e.onNext(1);
            System.out.println("1:" + System.currentTimeMillis());
            e.onNext(2);
            System.out.println("2:" + System.currentTimeMillis());
            e.onNext(3);
            System.out.println("3:" + System.currentTimeMillis());
            e.onComplete();
        });

        //观察者  消费会阻塞生产-----一对一队列消费？？？？
        Observer observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("======================onSubscribe");
            }

            @Override
            public void onNext(Integer integer) {
                System.out.println(("=========================currentThread name onNext: " + Thread.currentThread().getName()));
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(("======================onNext " + integer));
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("======================onError");
            }

            @Override
            public void onComplete() {
                System.out.println("======================onComplete");
            }
        };

        observable.subscribe(observer);
    }
}
